package com.xdxc.utils;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.xdxc.emuns.QuestionType;
import com.xdxc.mapper.TempUserAnswerMapper;
import com.xdxc.model.dto.UserAnswerDto;
import com.xdxc.model.entity.TempUserAnswer;
import com.xdxc.service.serviceinterface.TempUserAnswerService;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Supplier;
import java.util.stream.Collectors;

@Slf4j
@Component
public class MergeWindowManager {
    // 核心数据结构：线程安全的合并窗口
    private final ConcurrentHashMap<String, WindowBucket> windowMap = new ConcurrentHashMap<>();
    private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();

    // 配置参数（可通过配置中心动态调整）
    private static final long WINDOW_DURATION_MS = 6000; // 3秒合并窗口
    private static final int MAX_BATCH_SIZE = 20;       // 最大批量提交条数
    private static final int FLUSH_INTERVAL_MS = 8000;   // 刷写检查间隔



    @Autowired
    private TempUserAnswerService tempUserAnswerService;

    // 执行器服务
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(
            new ThreadFactoryBuilder().setNameFormat("MergeWindowManager-Flush-%d").build()
    );

    // 状态监控
    private final AtomicLong totalFlushed = new AtomicLong(0);
    private final AtomicLong lastFlushTime = new AtomicLong(0);
    private final AtomicInteger flushInProgress = new AtomicInteger(0);

    @Autowired
    private TempUserAnswerMapper userAnswerMapper;

    @PostConstruct
    public void init() {
        log.info("初始化合并窗口管理器[窗口时长={}ms, 刷出间隔={}ms]", WINDOW_DURATION_MS, FLUSH_INTERVAL_MS);
        scheduler.scheduleAtFixedRate(this::safeFlush,
                FLUSH_INTERVAL_MS, FLUSH_INTERVAL_MS, TimeUnit.MILLISECONDS);
    }

    @PreDestroy
    public void shutdown() {
        log.info("关闭合并窗口管理器");
        flushAllBuckets();
        scheduler.shutdown();
        try {
            if (!scheduler.awaitTermination(3, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 线程安全的累积方法
     */
    public void accumulate(UserAnswerDto dto) {
        Objects.requireNonNull(dto, "消息DTO不能为空");
        String compositeKey = buildCompositeKey(dto);

        rwLock.readLock().lock();
        try {
            windowMap.compute(compositeKey, (key, bucket) -> {
                if (bucket == null) {
                    log.debug("新建消息桶[key={}]", key);
                    return new WindowBucket(dto);
                }

                // 严格版本控制（使用消息自带的时间戳）
                if (dto.getBornTimestamp() > bucket.getLatestDTO().getBornTimestamp()) {
                    log.debug("更新消息桶[key={}]", key);
                    bucket.update(dto);
                } else {
                    log.debug("丢弃旧消息[key={}], 当前版本:{}, 收到版本:{}",
                            key, bucket.getLatestDTO().getBornTimestamp(), dto.getBornTimestamp());
                }
                return bucket;
            });
        } finally {
            rwLock.readLock().unlock();
        }

        checkBatchSize();
    }

    /**
     * 安全的刷出方法（定时任务调用）
     */
    private void safeFlush() {
        if (flushInProgress.get() > 0) {
            log.debug("已有刷出操作在进行中，跳过本次执行");
            return;
        }

        flushInProgress.incrementAndGet();
        try {
            long startTime = System.currentTimeMillis();
            log.debug("开始刷出检查[当前窗口大小={}]", windowMap.size());

            // 阶段1：收集过期桶
            List<WindowBucket> expiredBuckets = findExpiredBuckets(startTime);

            if (!expiredBuckets.isEmpty()) {
                // 阶段2：处理并提交数据
                processAndRemoveBuckets(expiredBuckets);

                // 记录状态
                lastFlushTime.set(startTime);
                totalFlushed.addAndGet(expiredBuckets.size());

                log.info("成功刷出{}条数据，窗口剩余{}",
                        expiredBuckets.size(), windowMap.size());
            }
        } catch (Exception e) {
            log.error("刷出操作异常", e);
        } finally {
            flushInProgress.decrementAndGet();
        }
    }

    /**
     * 查找过期桶（线程安全）
     */
    private List<WindowBucket> findExpiredBuckets(long currentTime) {
        List<WindowBucket> expiredBuckets = new ArrayList<>();

        windowMap.forEach((key, bucket) -> {
            synchronized (bucket) {
                if (currentTime - bucket.getLastUpdateTime() > WINDOW_DURATION_MS) {
                    expiredBuckets.add(bucket);
                    log.trace("标记过期桶[key={}], 最后更新时间:{}",
                            key, bucket.getLastUpdateTime());
                }
            }
        });

        return expiredBuckets;
    }

    /**
     * 处理并移除桶（原子性操作）
     */
    private void processAndRemoveBuckets(List<WindowBucket> buckets) {
        // 阶段1：准备批量数据
        List<TempUserAnswer> toFlush = new ArrayList<>(buckets.size());

        for (WindowBucket bucket : buckets) {
            synchronized (bucket) {
                TempUserAnswer entity = convertToEntity(bucket);
                toFlush.add(entity);
            }
        }

        // 阶段2：批量提交
        submitToBatchService(toFlush);

        // 阶段3：原子性移除
        for (WindowBucket bucket : buckets) {
            synchronized (bucket) {
                if (windowMap.get(bucket.getCompositeKey()) == bucket) {
                    windowMap.remove(bucket.getCompositeKey(), bucket);
                    log.trace("确认移除桶[key={}]", bucket.getCompositeKey());
                }
            }
        }
    }

    /**
     * 强制刷出所有数据
     */
    private void flushAllBuckets() {
        rwLock.writeLock().lock();
        try {
            // 调试点1：确认待处理数据量
            log.warn("‼️ 强制调试 - 当前窗口大小: {}", windowMap.size());

            List<WindowBucket> allBuckets = new ArrayList<>(windowMap.values());

            // 调试点2：确认实际获取的数据
            log.warn("‼️ 强制调试 - 待处理桶数量: {}", allBuckets.size());
            allBuckets.forEach(bucket ->
                    log.debug("待处理数据: {}", bucket.getLatestDTO()));

            windowMap.clear();

            List<TempUserAnswer> userAnswerList = allBuckets.stream()
                    .peek(bucket ->
                            log.trace("正在转换: {}", bucket.getCompositeKey()))
                    .map(this::convertToEntity)
                    .collect(Collectors.toList());

            // 调试点3：确认最终提交数据
            log.warn("‼️ 强制调试 - 准备提交数量: {}", userAnswerList.size());
            userAnswerList.forEach(e ->
                    log.trace("提交实体: {}", e.getId()));

            submitToBatchService(userAnswerList);
        } finally {
            rwLock.writeLock().unlock();
        }
    }

    /**
     * 批量大小检查
     */
    private void checkBatchSize() {
        if (windowMap.size() >= MAX_BATCH_SIZE) {
            log.info("触发批量大小阈值[当前大小={}]", windowMap.size());
            flushAllBuckets();
        }
    }

    /**
     * 提交到批量处理服务
     */
    private void submitToBatchService(List<TempUserAnswer> entities) {
        // 1. 防御性检查
        if (entities == null || entities.isEmpty()) {
            log.warn("⚠️ 提交空列表请求");
            return;
        }

        // 2. 记录原始数据指纹
        Map<Long, String> dataFingerprints = entities.stream()
                .collect(Collectors.toMap(
                        TempUserAnswer::getId,
                        e -> e.getAnswerText() + e.getAnswerIds()
                ));
        log.info("✅ 准备提交数据指纹: {}", dataFingerprints);

        try {
            // 3. 执行批量操作（带重试）
            executeWithRetry(() -> {
                tempUserAnswerService.safeBatchUpsert(entities);
                return null;
            });

            // 4. 提交后立即验证
            List<Long> savedIds = userAnswerMapper.selectList(
                    new QueryWrapper<TempUserAnswer>()
                            .in("id", dataFingerprints.keySet())
            ).stream().map(TempUserAnswer::getId).collect(Collectors.toList());

            // 5. 差异分析
//            if (savedIds.size() != entities.size()) {
//                Set<Long> lostIds = new HashSet<>(dataFingerprints.keySet());
//                lostIds.removeAll(savedIds);
//                log.error("❌ 数据丢失！缺失ID: {}", lostIds);
//                throw new RuntimeException("批量提交丢失了 " + lostIds.size() + " 条数据");
//            }

            log.info("✔️ 验证通过，全部{}条数据已持久化", entities.size());
        } catch (Exception e) {
            log.error("‼️ 批量提交失败，原始数据指纹: {}", dataFingerprints, e);
            throw e;
        }
    }
    private void executeWithRetry(Supplier<Void> operation) {
        int retry = 0;
        while (retry <= 3) {
            try {
                operation.get();
                return;
            } catch (DataAccessException e) {
                if (isDeadlock(e)) {
                    retry++;
                    log.warn("⚠️ 检测到死锁，第{}次重试...", retry);
                    sleepExponential(retry);
                } else {
                    throw e;
                }
            }
        }
        throw new RuntimeException("超过最大重试次数");
    }

    private boolean isDeadlock(DataAccessException e) {
        return e.getRootCause() != null
                && e.getRootCause().getMessage().contains("Deadlock");
    }

    private void sleepExponential(int retryCount) {
        try {
            Thread.sleep((long) (Math.pow(2, retryCount) * 100));
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
    }
    /**
     * 构建复合键：quId:userId:paperId
     */
    private String buildCompositeKey(UserAnswerDto dto) {
        return String.format("%d_%d_%d",
                dto.getQuId(),      // 题目ID
                dto.getUserId(),    // 用户ID
                dto.getPaperId()    // 试卷ID
        );
    }

    /**
     * 转换实体方法
     */
    private TempUserAnswer convertToEntity(WindowBucket bucket) {
        TempUserAnswer entity = new TempUserAnswer();
        UserAnswerDto dto = bucket.getLatestDTO();
        // 属性拷贝
        entity.setId(SnowflakeIdGenerator.nextId());
        entity.setUserId(dto.getUserId());
        entity.setQuId(dto.getQuId());
        entity.setPaperId(dto.getPaperId());
        entity.setAnswerType(QuestionType.getName(dto.getAnswerType()));
        entity.setAnswerText(dto.getAnswerText());
        entity.setAnswerIds(
                Optional.ofNullable(dto.getAnswerIds())
                        .orElse(Collections.emptyList())
                        .stream()
                        .map(Long::valueOf)
                        .collect(Collectors.toList())
        );
        entity.setVersion(dto.getVersion());
        entity.setExamId(dto.getExamId());
        entity.setCheckId(dto.getCheckId());
        return entity;
    }

    /**
     * 窗口桶数据结构（线程安全）
     */
    @Data
    private static class WindowBucket {
        private final String compositeKey;
        private volatile UserAnswerDto latestDTO;
        private final AtomicLong lastUpdateTime = new AtomicLong();
        private final AtomicInteger version = new AtomicInteger(0);

        public WindowBucket(UserAnswerDto dto) {
            this.compositeKey = dto.getCompositeKey();
            update(dto);
        }

        public void update(UserAnswerDto newDto) {
            this.latestDTO = newDto.deepCopy();
            this.lastUpdateTime.set(System.currentTimeMillis());
            this.version.incrementAndGet();
        }

        public long getLastUpdateTime() {
            return lastUpdateTime.get();
        }
    }

    /**
     * 监控端点
     */
    public Map<String, Object> getMetrics() {
        return Map.of(
                "windowSize", windowMap.size(),
                "totalFlushed", totalFlushed.get(),
                "lastFlushTime", new Date(lastFlushTime.get()),
                "flushInProgress", flushInProgress.get()
        );
    }
}